package com.atguigu.dga.meta.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.support.spring.PropertyPreFilters;
import com.atguigu.dga.meta.bean.TableMetaInfo;
import com.atguigu.dga.meta.bean.TableMetaInfoQuery;
import com.atguigu.dga.meta.bean.TableMetaInfoVO;
import com.atguigu.dga.meta.mapper.TableMetaInfoMapper;
import com.atguigu.dga.meta.service.TableMetaInfoExtraService;
import com.atguigu.dga.meta.service.TableMetaInfoService;
import com.atguigu.dga.util.SqlUtil;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
import org.apache.hadoop.hive.metastore.IMetaStoreClient;
import org.apache.hadoop.hive.metastore.api.MetaException;
import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
import org.apache.thrift.TException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

/**
 * <p>
 * 元数据表 服务实现类
 * </p>
 *
 * @author tubage
 * @since 2024-03-15
 */
@Service
@DS("dga")
public class TableMetaInfoServiceImpl extends ServiceImpl<TableMetaInfoMapper, TableMetaInfo> implements TableMetaInfoService {
    @Autowired
    private TableMetaInfoExtraService tableMetaInfoExtraService ;

    /**
     * 按照前端传入查询条件和分页信息 查询信息列表
     * @param tableMetaInfoQuery
     * @return
     */
    @Override
    public List<TableMetaInfoVO> getTableMetaInfoVOListByConditionAndPage(TableMetaInfoQuery tableMetaInfoQuery) {
        //因为条件是动态的， 需要在service中提前对sql进行处理
        StringBuilder sqlBuilder = new StringBuilder(
                "SELECT \n" +
                        "ti.id , " +
                        "ti.table_name, ti.schema_name, ti.table_size , ti.table_total_size , ti.table_comment ,\n" +
                        "ti.table_last_modify_time ,ti.table_last_access_time , \n" +
                        "te.tec_owner_user_name , te.busi_owner_user_name \n" +
                        "FROM table_meta_info ti \n" +
                        "JOIN  table_meta_info_extra te \n" +
                        "ON  ti.schema_name = te.schema_name AND ti.table_name = te.table_name \n" +
                        "WHERE ti.assess_date = ( SELECT MAX(assess_date) FROM table_meta_info ) "
        );

        // 条件: schemaName
        if(tableMetaInfoQuery.getSchemaName() != null && !tableMetaInfoQuery.getSchemaName().trim().isEmpty()){
            sqlBuilder.append(" AND ti.schema_name = '"+ SqlUtil.filterUnsafeSql( tableMetaInfoQuery.getSchemaName() ) +"'");
        }
        // 条件: tableName
        if(tableMetaInfoQuery.getTableName() != null && !tableMetaInfoQuery.getTableName().trim().isEmpty()){
            sqlBuilder.append(" AND ti.table_name like '%" + SqlUtil.filterUnsafeSql( tableMetaInfoQuery.getTableName())  + "%'" );
        }
        // 条件: dwLevel
        if(tableMetaInfoQuery.getDwLevel()!= null && !tableMetaInfoQuery.getDwLevel().trim().isEmpty()){
            sqlBuilder.append( " AND te.dw_level = '" + SqlUtil.filterUnsafeSql( tableMetaInfoQuery.getDwLevel() )+ "'");
        }

        //分页
        // 100条数据 , 每页显示10条数据
        // 1    0 - 9
        // 2   10 - 19
        // ....
        // 10  90 - 99

        // Limit : start ( ( pageNo - 1 ) * pageSize ) , size

        //计算开始行:
        int start = (tableMetaInfoQuery.getPageNo() - 1 ) * tableMetaInfoQuery.getPageSize()  ;

        sqlBuilder.append( " limit " + start +" , " + tableMetaInfoQuery.getPageSize()) ;

        //调用Mapper
        return getBaseMapper().selectTableMetaInfoVoList(sqlBuilder.toString());
    }

    // 列表结果总数
    @Override
    public Long getTableMetaInfoVOCountByCondition(TableMetaInfoQuery tableMetaInfoQuery) {
        StringBuilder sqlBuilder = new StringBuilder("select\n" +
                "    count(*) cnt\n" +
                "    from table_meta_info ti\n" +
                "join table_meta_info_extra te\n" +
                "on ti.schema_name = te.schema_name and ti.table_name = te.table_name\n" +
                "where assess_date = (select Max(assess_date) from table_meta_info)");

        // 条件: schemaName
        if(tableMetaInfoQuery.getSchemaName() != null && !tableMetaInfoQuery.getSchemaName().trim().isEmpty()){
            sqlBuilder.append(" AND ti.schema_name = '"+ SqlUtil.filterUnsafeSql( tableMetaInfoQuery.getSchemaName() ) +"'");
        }
        // 条件: tableName
        if(tableMetaInfoQuery.getTableName() != null && !tableMetaInfoQuery.getTableName().trim().isEmpty()){
            sqlBuilder.append(" AND ti.table_name like '%" + SqlUtil.filterUnsafeSql( tableMetaInfoQuery.getTableName())  + "%'" );
        }
        // 条件: dwLevel
        if(tableMetaInfoQuery.getDwLevel()!= null && !tableMetaInfoQuery.getDwLevel().trim().isEmpty()){
            sqlBuilder.append( " AND te.dw_level = '" + SqlUtil.filterUnsafeSql( tableMetaInfoQuery.getDwLevel() )+ "'");
        }

        return getBaseMapper().selectTableMetaInfoCount( sqlBuilder.toString());
    }

    @Override
    public void initTableMetaInfo(String schemaName, String assessDate) throws Exception {
        // TODO 0 先删除考评当天数据
        remove(
                new QueryWrapper<TableMetaInfo>().eq("assess_date", assessDate)
        );

        // TODO 1 从hive获取库中所有表
        List<String> allTables = hiveClient.getAllTables(schemaName);
        //        System.out.println(allTables);

        ArrayList<TableMetaInfo> tableMetaInfos = new ArrayList<>(allTables.size());

        // TODO 2 迭代处理每张表（保存到table_meta_info中（封装方法）、从HDFS获取信息、其他信息补充）
        for (String tableName : allTables) {
            Table table = hiveClient.getTable(schemaName, tableName);
//            System.out.println(table);
            TableMetaInfo tableMetaInfo = extractTableMetaInfoFromHive(table);
            extractTableMetaInfoFromHDFS(tableMetaInfo);
            tableMetaInfo.setAssessDate(assessDate);
            tableMetaInfo.setCreateTime(new Date()); // 修改时间用不上

            tableMetaInfos.add(tableMetaInfo);
        }
//        System.out.println(tableMetaInfos);
        // TODO 3 循环外批量导入
        saveBatch(tableMetaInfos);

        // TODO 4 extra 信息初始化
        tableMetaInfoExtraService.initTableMetaInfoExtra(tableMetaInfos);

    }

    // TODO 封装从hive获取信息方法: get set设置 返回table_meta_info
    private TableMetaInfo extractTableMetaInfoFromHive(Table table) {
        TableMetaInfo tableMetaInfo = new TableMetaInfo();

        tableMetaInfo.setSchemaName(table.getDbName());
        tableMetaInfo.setTableName(table.getTableName());
        PropertyPreFilters.MySimplePropertyPreFilter jsonFilter =
                new PropertyPreFilters().addFilter("comment", "name", "type");
        tableMetaInfo.setColNameJson(JSON.toJSONString(table.getSd().getCols(), jsonFilter)); // 所有列过滤后封装JSON
        tableMetaInfo.setPartitionColNameJson(JSON.toJSONString(table.getPartitionKeys(), jsonFilter));
        tableMetaInfo.setTableFsOwner( table.getOwner() );
        tableMetaInfo.setTableParametersJson(JSON.toJSONString(  table.getParameters() ) );
        tableMetaInfo.setTableComment( table.getParameters().get("comment"));
        tableMetaInfo.setTableFsPath( table.getSd().getLocation());
        tableMetaInfo.setTableInputFormat( table.getSd().getInputFormat());
        tableMetaInfo.setTableOutputFormat( table.getSd().getOutputFormat());
        tableMetaInfo.setTableRowFormatSerde( table.getSd().getSerdeInfo().getSerializationLib());
        tableMetaInfo.setTableCreateTime( new Date( table.getCreateTime() * 1000L)) ;
        tableMetaInfo.setTableType( table.getTableType() );
        //分桶数
        tableMetaInfo.setTableBucketNum((long) table.getSd().getNumBuckets());
        if(tableMetaInfo.getTableBucketNum() != -1L) {
            // 有分桶
            tableMetaInfo.setTableBucketColsJson(JSON.toJSONString(table.getSd().getBucketCols()));
            tableMetaInfo.setTableSortColsJson(JSON.toJSONString(table.getSd().getSortCols()));
        }
        return tableMetaInfo;
    }

    // TODO 封装从HDFS中提取信息 补充到table_meta_info中
    @Value("${hdfs.uris}")
    private String hdfsUris;
    private void extractTableMetaInfoFromHDFS(TableMetaInfo tableMetaInfo) throws Exception {
        // 获取文件对象、获取对应路径下所有表、递归、补充文件系统容量信息
        FileSystem fs = FileSystem.get(new URI(hdfsUris), new Configuration(), tableMetaInfo.getTableFsOwner());

        String tableFsPath = tableMetaInfo.getTableFsPath();
        tableFsPath = tableFsPath.replace("hadoop102:8020", "localhost:18020");
        FileStatus[] fileStatuses = fs.listStatus(new Path(tableFsPath));
//        System.out.println(fileStatuses);

        // 递归
        addHdfsInfoToTableMetaInfo(fs, fileStatuses, tableMetaInfo);

        //补充HDFS文件系统容量信息
        tableMetaInfo.setFsCapcitySize( fs.getStatus().getCapacity() );
        tableMetaInfo.setFsUsedSize( fs.getStatus().getUsed());
        tableMetaInfo.setFsRemainSize( fs.getStatus().getRemaining());

    }

    // TODO HDFS递归方法
    public void addHdfsInfoToTableMetaInfo(FileSystem fs, FileStatus[] fileStatuses, TableMetaInfo tableMetaInfo) throws IOException {
        //  循环处理路径下所有目录和文件
        for (FileStatus currFileStatus : fileStatuses) {
            if(currFileStatus.isFile()) {
                //表大小
                tableMetaInfo.setTableSize( ( tableMetaInfo.getTableSize()==null? 0L : tableMetaInfo.getTableSize() ) + currFileStatus.getLen()) ;
                //表总大小
                tableMetaInfo.setTableTotalSize( (tableMetaInfo.getTableTotalSize() == null ? 0L: tableMetaInfo.getTableTotalSize()) + currFileStatus.getLen() * currFileStatus.getReplication());
                //表的最后修改时间
                long modificationTime = currFileStatus.getModificationTime();
                long currModifyTime = tableMetaInfo.getTableLastModifyTime() == null ? 0L : tableMetaInfo.getTableLastModifyTime().getTime();
                tableMetaInfo.setTableLastModifyTime( new Date( Math.max( modificationTime , currModifyTime ) ));

                //表的最后访问时间
                long accessTime = currFileStatus.getAccessTime();
                long currAccessTime = tableMetaInfo.getTableLastAccessTime() == null ? 0L : tableMetaInfo.getTableLastAccessTime().getTime();
                tableMetaInfo.setTableLastAccessTime( new Date( Math.max( accessTime  , currAccessTime  ) ));
            } else {
                //下探
                //获取当前目录下所有的内容
                FileStatus[] subFileStatues = fs.listStatus(currFileStatus.getPath());
                addHdfsInfoToTableMetaInfo(fs , subFileStatues , tableMetaInfo) ;
            }
        }
    }


    //  TODO 0 创建hive客户端
    private IMetaStoreClient hiveClient;

    @Value("${hive.metastore.uris}")
    private String hiveMetastoreUris;

    @PostConstruct
    private void getHiveClient() {
        Configuration conf = new Configuration();
        MetastoreConf.setVar(conf, MetastoreConf.ConfVars.THRIFT_URIS, hiveMetastoreUris);

        try {
            hiveClient = new HiveMetaStoreClient(conf);
            System.out.println(hiveClient);
        } catch (MetaException e) {
            throw new RuntimeException("获取hive客户端失败");
        }
    }
}
